//package com.raven.play.stream.test.javadsl;
//
//import akka.Done;
//import akka.NotUsed;
//import akka.actor.ActorSystem;
//import akka.stream.alpakka.slick.javadsl.Slick;
//import akka.stream.alpakka.slick.javadsl.SlickSession;
//import akka.stream.javadsl.RunnableGraph;
//import akka.stream.javadsl.Sink;
//import akka.stream.javadsl.Source;
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;
//import scala.collection.immutable.Vector;
//import scala.collection.mutable.ReusableBuilder;
//import scala.concurrent.Await;
//import scala.concurrent.Future;
//import scala.concurrent.duration.Duration;
//
//import java.sql.PreparedStatement;
//import java.util.List;
//import java.util.concurrent.CompletionStage;
//import java.util.concurrent.TimeUnit;
//import java.util.function.BiConsumer;
//import java.util.stream.IntStream;
//
//record User(Integer id, String name) {
//}
//
//public class JdbcTest1 {
//
//    static Logger logger = LoggerFactory.getLogger(JdbcTest1.class);
//
//    public static void main(String[] args) {
//        ActorSystem system = ActorSystem.create("slick-example");
//        // final Materializer materializer = Materializer.createMaterializer(system);
//
//
//        SlickSession session = SlickHelper.createSession("slick-h2");
//        system.registerOnTermination(session::close);
//
//        String createTableStatement = "CREATE TABLE IF NOT EXISTS AKKA_TEST_USER (id INT PRIMARY KEY, name VARCHAR(255))";
//        // Seq<String> statementSeq = JavaConverters.asScalaBufferConverter(Collections.singletonList(createTableStatement)).asScala().toSeq();
//        // SQLActionBuilder sqlActionBuilder = new SQLActionBuilder(statementSeq, SetParameter.SetUnit$.MODULE$);
//        // SqlAction<Integer, NoStream, Effect.Write> createAction = sqlActionBuilder.asUpdate();
//        // session.db().run(createAction);
//
//        // ScalaSlickHelper.runSql(createTableStatement, session);
//        Future<Object> createTableDone = session.db().run(ScalaSlickHelper.createSqlAction(createTableStatement));
//
//        try {
//            Await.result(createTableDone, Duration.create(30, TimeUnit.SECONDS));
//        } catch (Exception e) {
//            logger.error(e.getMessage(), e);
//            system.terminate();
//            throw new RuntimeException(e);
//        }
//
//        String truncateTableStatement = "TRUNCATE TABLE AKKA_TEST_USER";
//        Future<Object> truncateTableDone = session.db().run(ScalaSlickHelper.createSqlAction(truncateTableStatement));
//
//        try {
//            Await.result(truncateTableDone, Duration.create(30, TimeUnit.SECONDS));
//        } catch (Exception e) {
//            logger.error(e.getMessage(), e);
//            system.terminate();
//            throw new RuntimeException(e);
//        }
//
//        List<User> users = IntStream.range(4, 7)
//                .boxed()
//                .map(i -> new User(i, "Name" + i))
//                .toList();
//
//        Sink<User, CompletionStage<Done>> sink = Slick.<User>sink(
//                session,
//                (user, connection) -> {
//                    PreparedStatement statement = connection.prepareStatement("INSERT INTO AKKA_TEST_USER VALUES (?,?)");
//                    statement.setInt(1, user.id());
//                    statement.setString(2, user.name());
//                    return statement;
//                }
//        );
//
//        CompletionStage<Done> insertDone = Source.from(users)
//                .alsoTo(Sink.foreach(it -> logger.info(it.toString())))
//                .runWith(sink, system);
//
//        RunnableGraph<NotUsed> graph = Source.from(users).to(sink);
//        graph.run(system);
//
//        try {
//            insertDone.toCompletableFuture().get(30, TimeUnit.SECONDS);
//        } catch (Exception e) {
//            logger.error(e.getMessage(), e);
//            system.terminate();
//            throw new RuntimeException(e);
//        }
//
//        CompletionStage<Done> selectDone = Slick.source(session,
//                        "SELECT ID, NAME FROM AKKA_TEST_USER",
//                        slickRow -> {
//                            ReusableBuilder<Object, Vector<Object>> vectorReusableBuilder = Vector.newBuilder();
//                            vectorReusableBuilder.addOne(slickRow.nextInt()).addOne(slickRow.nextString());
//                            return vectorReusableBuilder.result();
//                        })
//                .runForeach(o -> logger.info(o.toString()), system);
//
//        selectDone.whenComplete((done, exception) -> {
//            if (exception != null) {
//                logger.error("流处理失败: " + exception.getMessage());
//            } else {
//                logger.info("流处理成功");
//            }
//            system.terminate();
//        });
//
//        try {
//            selectDone.toCompletableFuture().get();
//        } catch (Exception e) {
//            logger.error(e.getMessage(), e);
//            system.terminate();
//            throw new RuntimeException(e);
//        }
//    }
//}
